package com.dddframework.mq.impl;

import cn.hutool.core.thread.GlobalThreadPool;
import com.dddframework.core.contract.MQEvent;
import com.dddframework.core.contract.exception.ServiceException;
import com.dddframework.core.utils.JsonKit;
import com.dddframework.mq.core.MQClient;
import com.dddframework.mq.core.MQFilter;
import com.dddframework.mq.core.MQListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;

@Component
@Slf4j(topic = "### BASE-MQ : rabbitClient ###")
public final class RabbitClient implements MQClient {
    private static Connection CONNECTION;

    @Override
    public String impl() {
        return "rabbit";
    }

    @Override
    public Consumer<MQEvent> initProducer() {
        try {
            com.rabbitmq.client.Channel channel = connection().createChannel();
            return mqEvent -> {
                // 定义具体的MQ事件发布逻辑
                String message = serialization().serialize(mqEvent);
                log.info("Publish MQ: {}", message);
                try {
                    // 路由键=namespace.topic或namespace.topic.tag
                    String routingKey = config().namespace(".") + mqEvent.getTopic() + (mqEvent.getTag() == null || mqEvent.getTag().isEmpty() ? "" : ("." + mqEvent.getTag()));
                    channel.basicPublish(config().getExchange(), routingKey, null, message.getBytes());
                } catch (Exception e) {
                    log.error("Publish MQ: {} failed!", message, e);
                }
            };
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public boolean initConsumer(MQListener mqListener) throws Exception {
        // 队列名=group.namespace.topic.className.methodName
        String queue = mqListener.group(".") + config().namespace(".") + mqListener.getMethod().getDeclaringClass().getSimpleName() + "." + mqListener.getMethod().getName();
        List<String> routingKeys = new ArrayList<>();
        if (mqListener.getTags() != null && !mqListener.getTags().isEmpty()) {
            // 通过多个tags绑定到对应的路由键
            Set<String> tags = MQFilter.findIncludes(mqListener.getTags());
            if (!tags.isEmpty()) {
                // 路由键=namespace.topic.tag1、namespace.topic.tag2，TODO 这里只能用或||的关系，通配符*和非-是不能生效的
                for (String tag : tags) {
                    String routingKey = config().namespace(".") + mqListener.getTopic() + "." + tag;
                    routingKeys.add(routingKey);
                }
            }
        } else {
            // 路由键=namespace.topic
            String routingKey = config().namespace(".") + mqListener.getTopic();
            routingKeys.add(routingKey);
        }
        try (com.rabbitmq.client.Channel channel = connection().createChannel()) {
            channel.queueDeclare(queue, true, false, false, null);
            // 队列绑定到多个路由器
            for (String routingKey : routingKeys) {
                channel.queueBind(queue, config().getExchange(), routingKey);
            }
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                MQEvent mqEvent = mqListener.getDeserialize().apply(message);
                if (!MQFilter.matchExps(mqEvent.getTag(), mqListener.getTags())) {
                    return;
                }
                try {
                    consume(mqListener, mqEvent);
                    if (!config().isAutoAck()) {
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    }
                } catch (Exception e) {
                    if (e instanceof ServiceException) {
                        log.error("Consume MQ failed: {}. event: {}, ", e.getMessage(), JsonKit.toJson(mqEvent));
                        if (!config().isAutoAck()) {
                            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                        }
                    } else {
                        log.error("Consume MQ failed: {}", mqEvent, e);
                    }
                }
            };
            GlobalThreadPool.submit(() -> {
                try {
                    channel.basicConsume(queue, config().isAutoAck(), deliverCallback, consumerTag -> {
                    });
                } catch (Exception e) {
                }
                return channel;
            });
        }
        return true;
    }

    private Connection connection() {
        if (CONNECTION == null) {
            ConnectionFactory factory = new ConnectionFactory();
            String[] hostAndPort = config().getServer().split(":");
            factory.setHost(hostAndPort[0]);
            factory.setPort(Integer.parseInt(hostAndPort[1]));
            factory.setUsername(config().getUsername());
            factory.setPassword(config().getPassword());
            try {
                CONNECTION = factory.newConnection();
            } catch (IOException | TimeoutException e) {
                throw new RuntimeException(e);
            }
        }
        return CONNECTION;
    }
}